package com.sui.bigdata.flink.sql.launcher.yarn;

import com.sui.bigdata.flink.sql.launcher.ClusterClientFactory;
import com.sun.istack.Nullable;
import org.apache.hadoop.yarn.api.records.ApplicationSubmissionContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.Set;

public class ApplicationSubmissionContextReflector {
	private static Logger LOG = LoggerFactory.getLogger(ClusterClientFactory.class);

	private static final ApplicationSubmissionContextReflector instance =
			new ApplicationSubmissionContextReflector(ApplicationSubmissionContext.class);

	public static ApplicationSubmissionContextReflector getInstance() {
		return instance;
	}

	private static final String APPLICATION_TAGS_METHOD_NAME = "setApplicationTags";
	private static final String ATTEMPT_FAILURES_METHOD_NAME = "setAttemptFailuresValidityInterval";
	private static final String KEEP_CONTAINERS_METHOD_NAME = "setKeepContainersAcrossApplicationAttempts";
	private static final String NODE_LABEL_EXPRESSION_NAME = "setNodeLabelExpression";

	private final Method applicationTagsMethod;
	private final Method attemptFailuresValidityIntervalMethod;
	private final Method keepContainersMethod;
	@Nullable
	private final Method nodeLabelExpressionMethod;

	private ApplicationSubmissionContextReflector(Class<ApplicationSubmissionContext> clazz) {
		Method applicationTagsMethod;
		Method attemptFailuresValidityIntervalMethod;
		Method keepContainersMethod;
		Method nodeLabelExpressionMethod;

		try {
			// this method is only supported by Hadoop 2.4.0 onwards
			applicationTagsMethod = clazz.getMethod(APPLICATION_TAGS_METHOD_NAME, Set.class);
			LOG.debug("{} supports method {}.", clazz.getCanonicalName(), APPLICATION_TAGS_METHOD_NAME);
		} catch (NoSuchMethodException e) {
			LOG.debug("{} does not support method {}.", clazz.getCanonicalName(), APPLICATION_TAGS_METHOD_NAME);
			// assign null because the Hadoop version apparently does not support this call.
			applicationTagsMethod = null;
		}

		this.applicationTagsMethod = applicationTagsMethod;

		try {
			// this method is only supported by Hadoop 2.6.0 onwards
			attemptFailuresValidityIntervalMethod = clazz.getMethod(ATTEMPT_FAILURES_METHOD_NAME, long.class);
			LOG.debug("{} supports method {}.", clazz.getCanonicalName(), ATTEMPT_FAILURES_METHOD_NAME);
		} catch (NoSuchMethodException e) {
			LOG.debug("{} does not support method {}.", clazz.getCanonicalName(), ATTEMPT_FAILURES_METHOD_NAME);
			// assign null because the Hadoop version apparently does not support this call.
			attemptFailuresValidityIntervalMethod = null;
		}

		this.attemptFailuresValidityIntervalMethod = attemptFailuresValidityIntervalMethod;

		try {
			// this method is only supported by Hadoop 2.4.0 onwards
			keepContainersMethod = clazz.getMethod(KEEP_CONTAINERS_METHOD_NAME, boolean.class);
			LOG.debug("{} supports method {}.", clazz.getCanonicalName(), KEEP_CONTAINERS_METHOD_NAME);
		} catch (NoSuchMethodException e) {
			LOG.debug("{} does not support method {}.", clazz.getCanonicalName(), KEEP_CONTAINERS_METHOD_NAME);
			// assign null because the Hadoop version apparently does not support this call.
			keepContainersMethod = null;
		}

		this.keepContainersMethod = keepContainersMethod;

		try {
			nodeLabelExpressionMethod = clazz.getMethod(NODE_LABEL_EXPRESSION_NAME, String.class);
			LOG.debug("{} supports method {}.", clazz.getCanonicalName(), NODE_LABEL_EXPRESSION_NAME);
		} catch (NoSuchMethodException e) {
			LOG.debug("{} does not support method {}.", clazz.getCanonicalName(), NODE_LABEL_EXPRESSION_NAME);
			nodeLabelExpressionMethod = null;
		}

		this.nodeLabelExpressionMethod = nodeLabelExpressionMethod;
	}
	public void setApplicationTags(
			ApplicationSubmissionContext appContext,
			Set<String> applicationTags) throws InvocationTargetException, IllegalAccessException {
		if (applicationTagsMethod != null) {
			LOG.debug("Calling method {} of {}.",
					applicationTagsMethod.getName(),
					appContext.getClass().getCanonicalName());
			applicationTagsMethod.invoke(appContext, applicationTags);
		} else {
			LOG.debug("{} does not support method {}. Doing nothing.",
					appContext.getClass().getCanonicalName(),
					APPLICATION_TAGS_METHOD_NAME);
		}
	}

	public void setApplicationNodeLabel(
			ApplicationSubmissionContext appContext,
			String nodeLabel) throws InvocationTargetException, IllegalAccessException {
		if (nodeLabelExpressionMethod != null) {
			LOG.debug("Calling method {} of {}.",
					nodeLabelExpressionMethod.getName(),
					appContext.getClass().getCanonicalName());
			nodeLabelExpressionMethod.invoke(appContext, nodeLabel);
		} else {
			LOG.debug("{} does not support method {}. Doing nothing.",
					appContext.getClass().getCanonicalName(),
					NODE_LABEL_EXPRESSION_NAME);
		}
	}

	public void setAttemptFailuresValidityInterval(
			ApplicationSubmissionContext appContext,
			long validityInterval) throws InvocationTargetException, IllegalAccessException {
		if (attemptFailuresValidityIntervalMethod != null) {
			LOG.debug("Calling method {} of {}.",
					attemptFailuresValidityIntervalMethod.getName(),
					appContext.getClass().getCanonicalName());
			attemptFailuresValidityIntervalMethod.invoke(appContext, validityInterval);
		} else {
			LOG.debug("{} does not support method {}. Doing nothing.",
					appContext.getClass().getCanonicalName(),
					ATTEMPT_FAILURES_METHOD_NAME);
		}
	}

	public void setKeepContainersAcrossApplicationAttempts(
			ApplicationSubmissionContext appContext,
			boolean keepContainers) throws InvocationTargetException, IllegalAccessException {

		if (keepContainersMethod != null) {
			LOG.debug("Calling method {} of {}.", keepContainersMethod.getName(),
					appContext.getClass().getCanonicalName());
			keepContainersMethod.invoke(appContext, keepContainers);
		} else {
			LOG.debug("{} does not support method {}. Doing nothing.",
					appContext.getClass().getCanonicalName(), KEEP_CONTAINERS_METHOD_NAME);
		}
	}
}